package com.lrj.datastream.transform.basic;

import com.lrj.datastream.domain.ClickEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author lrj
 * @date 2022/3/28 15:24
 */
public class TestMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        final DataStreamSource<ClickEvent> stream = env.fromElements(new ClickEvent("Mary", "./home", 1000L),
                new ClickEvent("Blob", "./home", 2000L),
                new ClickEvent("Mary", "./home", 3000L));
        // 1.自定义MapFunction
        stream.map(new MyMapper()).print();
        // 2.匿名内部类
        stream.map(new MapFunction<ClickEvent, String>() {
            @Override
            public String map(ClickEvent value) throws Exception {
                return value.getUrl();
            }
        }).print();
        // 4.lamda,泛型擦除,需要注意returns
        stream.map(ClickEvent::getUser).print();
        env.execute(TestMap.class.getName());
    }

    private static class MyMapper implements MapFunction<ClickEvent, String> {
        @Override
        public String map(ClickEvent value) throws Exception {
            return value.getUser();
        }
    }
}
